package com.atguigu.tingshu.album.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.conn.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.AuthContextHolder;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.micrometer.core.instrument.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.util.descriptor.web.ContextHandler;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.net.ContentHandler;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
@Transactional
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;
    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private RedissonClient redissonClient;


    //添加专辑
    @Override
    public void saveAlbumInfo(AlbumInfoVo albumInfoVo) {

        //1.专辑基本信息表
        AlbumInfo albumInfo = new AlbumInfo();
        BeanUtils.copyProperties(albumInfoVo, albumInfo);

        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);

        //获取userId
        Long userId = AuthContextHolder.getUserId();
        albumInfo.setUserId(userId);
        //如果不是免费的,那么只能试听5集
        if (!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfoVo.getPayType())) {
            albumInfo.setTracksForFree(5);
        }
        albumInfoMapper.insert(albumInfo);


        //2.专辑属性值表
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (!StringUtils.isEmpty(albumAttributeValueVoList)) {
            albumAttributeValueVoList.stream().forEach(albumAttributeValueVo -> {
                        AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                        BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                        albumAttributeValue.setAlbumId(albumInfo.getId());
                        albumAttributeValueMapper.insert(albumAttributeValue);
                    }
            );
        }

        //3.统计数据
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);

        //通过专辑的isOpen 判断是否公开,需要将专辑上架或者下架
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, String.valueOf(albumInfo.getId()));
        }

    }

    //分页查询用户专辑
    @Override
    public IPage<AlbumListVo> findUserAlbumPage(Long page, Long limit, AlbumListVo albumListVo) {
        Page<AlbumListVo> result = new Page<>(page, limit);
        return albumInfoMapper.findUserAlbumPage(result, albumListVo);
    }

    //根据专辑id删除专辑
    @Override
    public void removeAlbumInfo(Long id) {
        //1.删除专辑基本信息表
        LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumInfoLambdaQueryWrapper.eq(AlbumInfo::getId, id);
        albumInfoMapper.delete(albumInfoLambdaQueryWrapper);
        //2.删除专辑属性值表
        LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);
        //3.删除统计数据
        LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
        albumStatMapper.delete(albumStatLambdaQueryWrapper);

        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(id));
    }


    //根据id查询专辑基本信息
    @Override
    @GuiGuCache(cacheKey = RedisConstant.ALBUM_INFO_PREFIX+"#{#params[0]}",
                 enableLock = true,lockName =RedisConstant.ALBUM_LOCK_SUFFIX +"#{#params[0]}" )
    public AlbumInfo getAlbumInfo(Long id) {

        return getAlbumInfoFromDataBase(id);
    }

    //根据id查询专辑基本信息
    public AlbumInfo getAlbumInfocount(Long id) {

        //1.先从redis中查询专辑数据
        String redisAlbumInfoJSON = redisTemplate.opsForValue().get(RedisConstant.ALBUM_INFO_PREFIX + id);
        if (!StringUtils.isEmpty(redisAlbumInfoJSON)) {
            return JSON.parseObject(redisAlbumInfoJSON, AlbumInfo.class);
        }

        //如果redis中没有获取到数据,那么就使用分布式锁来解决缓存击穿的问题
        //获取分布式锁
        RLock lock = redissonClient.getLock(RedisConstant.ALBUM_LOCK_SUFFIX + id);
        boolean tryLock = lock.tryLock();
        if (tryLock) {
            log.info("线程{}获取到了锁", Thread.currentThread().getName());
            try {
                //从数据库中查询获取专辑数据
                AlbumInfo albumInfo = getAlbumInfoFromDataBase(id);
                //这里不对这个albumInfo进行判空校验,因为我们使用了缓存,即使前端发送了一个不存在的专辑id,
                // 那么我们在redis中也把他设置为null,这样就初步的解决了缓存穿透的问题

                //然后存储到redis中
                redisTemplate.opsForValue().set(RedisConstant.ALBUM_INFO_PREFIX + id, JSON.toJSONString(albumInfo),
                        RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);

                return albumInfo;
            }catch (Exception e){
                throw new RuntimeException("获取锁失败");
            }finally {
                //释放锁
                lock.unlock();
                log.info("线程{}释放了锁", Thread.currentThread().getName());
            }
        }
        else {
            log.info("线程{}没有获取到锁,进行自旋", Thread.currentThread().getName());
            return getAlbumInfo(id);
        }
    }

    private AlbumInfo getAlbumInfoFromDataBase(Long id) {
        AlbumInfo albumInfo = this.getById(id);
        if (albumInfo != null) {
            //根据id查询专辑属性值
            LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
            albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
           // albumAttributeValueLambdaQueryWrapper.select(AlbumAttributeValue::getAlbumId,id,AlbumAttributeValue::getAttributeId,)
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(albumAttributeValueLambdaQueryWrapper);
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
        }
        return albumInfo;
    }

    //根据专辑id修改专辑信息
    @Override
    public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {


        //保证更新时缓存数据和数据库数据一致,避免数据不一致问题
        //采用延时双删
        //redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX+albumId);



        //1.修改专辑基本信息表
        //先要查询原数据,在原数据的基础上进行修改
        AlbumInfo albumInfo = albumInfoMapper.selectById(albumId);
        BeanUtils.copyProperties(albumInfoVo, albumInfo);
        albumInfoMapper.updateById(albumInfo);

        //2.先删除专辑属性值表
        LambdaQueryWrapper<AlbumAttributeValue> attributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        attributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, albumId);
        albumAttributeValueMapper.delete(attributeValueLambdaQueryWrapper);

        //3.再添加专辑属性值表
        if (!StringUtils.isEmpty(albumInfoVo.getAlbumAttributeValueVoList())) {
            albumInfoVo.getAlbumAttributeValueVoList().forEach(albumAttributeValueVo -> {
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                albumAttributeValue.setAlbumId(albumId);
                albumAttributeValueMapper.insert(albumAttributeValue);
            });
        }



//        try {
//            //延迟休眠,休眠时间估值
//            TimeUnit.MILLISECONDS.sleep(200);
//
//            //进行第二次删除
//            redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX+albumId);
//
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }



        //通过专辑的isOpen 判断是否公开,需要将专辑上架或者下架
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, String.valueOf(albumInfo.getId()));
        } else {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(albumInfo.getId()));
        }


    }


    //获取当前用户全部专辑列表
    @Override
    public List<AlbumInfo> findUserAllAlbumList() {
        Long userId = AuthContextHolder.getUserId();
        LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumInfoLambdaQueryWrapper.eq(AlbumInfo::getUserId, userId)
                .select(AlbumInfo::getAlbumTitle, AlbumInfo::getId)
                .orderByDesc(AlbumInfo::getId);
        return albumInfoMapper.selectList(albumInfoLambdaQueryWrapper);

    }

    @GuiGuCache(cacheKey = "album:stat:" + "#{#params[0]}")
    @Override
    public AlbumStatVo findAlbumStatByAlbumId(Long albumId) {
        return albumStatMapper.findAlbumStatByAlbumId(albumId);
    }


    public void saveAlbumStat(Long albumId, String statType) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(0);
        albumStatMapper.insert(albumStat);
    }
}
